package com.atguigu.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.Func.MyWindowFunction;
import com.atguigu.bean.PageViewBean;
import com.atguigu.util.ClickHouseUtil_hjy;
import com.atguigu.util.DateFormatUtil_hjy;
import com.atguigu.util.KafkaUtil_wm;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Dws_PageViewWindow {
    public static void main(String[] args) throws Exception {
        // TODO: 2023/3/25 需求：页面浏览窗口汇总表
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism( 1 );
        // TODO: 2023/3/25 读取kafka主题 流量域事实表 dwd_traffic_page_log 主题数据
        String topic = "dwd_traffic_page_log";
        String groupId = "Dws_PageViewWindow_11";
        DataStreamSource<String> kafkaDS = env.addSource( KafkaUtil_wm.getFlinkKafkaConsumer( topic, groupId ) );
        // TODO: 2023/3/25 过滤数据并转为json格式
        SingleOutputStreamOperator<JSONObject> mapDS = kafkaDS.map( new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject( value );
                return jsonObject;
            }
        } );
//        mapDS.print( "mapDS>>>>>>>>>>>>>>>>>>>>>" );
        SingleOutputStreamOperator<JSONObject> jsonObjDS = mapDS.filter( new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String pageId = value.getJSONObject( "page" ).getString( "page_id" );

                return "home".equals( pageId ) || "good_detail".equals( pageId );
            }
        } );
//        jsonObjDS.print( "jsonObjDS>>>>>>>>>>>>>>" );

        // TODO: 2023/3/25 提取时间戳
        SingleOutputStreamOperator<JSONObject> jsonDSWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<JSONObject>forBoundedOutOfOrderness( Duration.ofSeconds( 5 ) )
                        .withTimestampAssigner( new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                return element.getLong( "ts" );
                            }
                        } )
        );
        jsonDSWithWMDS.print( "jsonDSWithWMDS>>>>>>>>>>>>>>>>>>>>" );
        // TODO: 2023/3/25 按照mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonDSWithWMDS.keyBy( line -> line.getJSONObject( "common" ).getString( "mid" ) );
        // TODO: 2023/3/25 状态编程  并转为JavaBean对象
        SingleOutputStreamOperator<PageViewBean> trafficHomeDetailPageViewDS = keyedStream.flatMap( new RichFlatMapFunction<JSONObject, PageViewBean>() {
            private ValueState<String> lasthomeState;
            private ValueState<String> lastgooddetailState;

            @Override
            public void open(Configuration parameters) throws Exception {
                StateTtlConfig ttlConfig = new StateTtlConfig.Builder( Time.days( 1 ) )
                        .setUpdateType( StateTtlConfig.UpdateType.OnReadAndWrite )
                        .build();
                ValueStateDescriptor<String> homeState = new ValueStateDescriptor<>( "home-state", String.class );
                ValueStateDescriptor<String> gooddetailState = new ValueStateDescriptor<>( "gooddetail-state", String.class );
                homeState.enableTimeToLive( ttlConfig );
                gooddetailState.enableTimeToLive( ttlConfig );
                lasthomeState = getRuntimeContext().getState( homeState );
                lastgooddetailState = getRuntimeContext().getState( gooddetailState );
            }

            @Override
            public void flatMap(JSONObject value, Collector<PageViewBean> out) throws Exception {
                String lastHomeDt = lasthomeState.value();
                String lastGoodDetailDt = lastgooddetailState.value();
                String curDt = DateFormatUtil_hjy.toDate( value.getLong( "ts" ) );
                String pageId = value.getJSONObject( "page" ).getString( "page_id" );
                // TODO: 2023/3/25 如果是home，否则就是商品详情页
                long homeCount = 0L;
                long gooddetailCount = 0L;
                if ("home".equals( pageId )) {
                    if (lastHomeDt == null || lastHomeDt.compareTo( curDt ) < 0) {
                        homeCount = 1L;
                        lasthomeState.update( curDt );
                    }
                } else {
                    if (lastGoodDetailDt == null || lastGoodDetailDt.compareTo( curDt ) < 0) {
                        gooddetailCount = 1L;
                        lastgooddetailState.update( curDt );
                    }
                }
                if (homeCount != 0 || gooddetailCount != 0) {
                    out.collect( new PageViewBean(
                                    "", "",
                                    homeCount,
                                    gooddetailCount,
                                    null
                            )

                    );
                }
            }
        } );
        trafficHomeDetailPageViewDS.print( "trafficHomeDetailPageViewDS>>>>>>>>>>>>>>>>>>>" );
        // TODO: 2023/3/25 开窗聚合
        SingleOutputStreamOperator<PageViewBean> redultDS = trafficHomeDetailPageViewDS.windowAll( TumblingProcessingTimeWindows.of( org.apache.flink.streaming.api.windowing.time.Time.seconds( 10 ) ) )
                .reduce( new ReduceFunction<PageViewBean>() {
                             @Override
                             public PageViewBean reduce(PageViewBean value1, PageViewBean value2) throws Exception {
                                 value1.setHomeUvCt( value1.getHomeUvCt() + value2.getHomeUvCt() );
                                 value1.setGoodDetailUvCt( value1.getGoodDetailUvCt() + value2.getGoodDetailUvCt() );
                                 return value1;
                             }
                         },
                        new AllWindowFunction<PageViewBean, PageViewBean, TimeWindow>() {
                            @Override
                            public void apply(TimeWindow window, Iterable<PageViewBean> values, Collector<PageViewBean> out) throws Exception {
                                PageViewBean next = values.iterator().next();
                                out.collect( MyWindowFunction.getWindow( next, window ) );
                            }
                        } );
        // TODO: 2023/3/25 启动任务
        redultDS.print( "redultDS>>>>>>>>>>>>>>>>>>>>>" );
        redultDS.addSink( ClickHouseUtil_hjy.getClickHouseSink( "insert into dws_traffic_page_view_window values(?,?,?,?,?)" ) );
        env.execute();
    }
}
